package cn.sciento.boot.transfer.app.service.impl;

import cn.sciento.core.exception.CommonException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import cn.sciento.boot.data.change.dto.ChangeData;
import cn.sciento.boot.transfer.api.dto.EventErrorDTO;
import cn.sciento.boot.transfer.api.dto.EventSendDTO;
import cn.sciento.boot.transfer.api.dto.ProcessMsgDTO;
import cn.sciento.boot.transfer.api.dto.ProducerConfigDTO;
import cn.sciento.boot.transfer.app.service.ProducerEventArchService;
import cn.sciento.boot.transfer.app.service.ProducerEventService;
import cn.sciento.boot.transfer.domain.entity.ProducerEvent;
import cn.sciento.boot.transfer.domain.repository.ProducerEventArchRepository;
import cn.sciento.boot.transfer.domain.repository.ProducerEventRepository;
import cn.sciento.boot.transfer.infra.feign.HdttRemoteService;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.core.redis.RedisHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class ProducerEventServiceImpl implements ProducerEventService {
  private Logger LOGGER = LoggerFactory.getLogger(ProducerEventServiceImpl.class);
  
  @Value("${spring.application.name}")
  private String serviceName;
  
  @Autowired
  private ProducerEventRepository producerEventRepository;
  
  @Autowired
  private ProducerEventArchRepository producerEventArchRepository;
  
  @Autowired
  private ObjectMapper objectMapper;
  
  @Autowired
  private ProducerEventArchService producerEventArchService;
  
  @Autowired
  private RedisHelper redisHelper;
  
  @Autowired
  private HdttRemoteService hdttRemoteService;
  
  @Transactional(rollbackFor = {Exception.class})
  public List<ProducerEvent> addProducerEvent(List<ChangeData> changeDataList) {
    if (CollectionUtils.isEmpty(changeDataList)) {
      this.LOGGER.info("changeData list is null");
      return null;
    } 
    ChangeData changeData = changeDataList.get(0);
    ProducerConfigDTO producerConfig = getProducerConfig(changeData.getTableName());
    if(producerConfig  == null){
      return new ArrayList<>();
    }
    Integer tenantFlag = producerConfig.getTenantFlag();
    this.LOGGER.info("start to build processMsg list");
    List<ProcessMsgDTO> processMsgDTOS = ProcessMsgDTO.buildProcessMsgList(this.hdttRemoteService, changeDataList, this.serviceName, tenantFlag);
    List<ProducerEvent> producerEvents = new ArrayList<>();
    for (ProcessMsgDTO processMsgDTO : processMsgDTOS) {
      try {
        String processMsgStr = this.objectMapper.writeValueAsString(processMsgDTO);
        ProducerEvent producerEvent = ProducerEvent.buildProducerEvent(processMsgDTO, processMsgStr);
        producerEvents.add(producerEvent);
      } catch (JsonProcessingException e) {
        throw new CommonException("write value as string error");
      } 
    } 
    this.LOGGER.info("start to insert producerEvents");
    this.producerEventRepository.batchInsertSelective(producerEvents);
    return producerEvents;
  }
  
  public Boolean existProducerConfig(String tableName) {
    ProducerConfigDTO producerConfig = getProducerConfig(tableName);
    if (null == producerConfig)
      return Boolean.FALSE;
    if (BaseConstants.Flag.NO.equals(producerConfig.getEnabledFlag()))
      return Boolean.FALSE;
    return Boolean.TRUE;
  }
  
  @Async("stong-boot-transfer-executor")
  public void sentProducerEvent(ProducerEvent producerEvent) {
    this.LOGGER.info("start to sent producer event");
    if (null == producerEvent) {
      this.LOGGER.info("producerEvent is null");
      return;
    } 
    producerEvent.setProcessStatus("R");
    EventSendDTO eventSendDTO = this.hdttRemoteService.createEventSend(EventSendDTO.createInstance(producerEvent));
    if (null == eventSendDTO || null == eventSendDTO.getEventId()) {
      this.LOGGER.info("remote sent producer event failed");
      EventErrorDTO eventErrorDTO = this.hdttRemoteService.createEventError(EventErrorDTO.getInstance(producerEvent, "SYNC", "producer Event create fail"));
      producerEvent.setProcessStatus("ER");
      if (null == eventErrorDTO || null == eventErrorDTO.getEventId())
        producerEvent.setProcessStatus("ERE"); 
      this.producerEventRepository.updateOptional(producerEvent, "processStatus");
    } else {
      try {
        this.LOGGER.info("start to create producer eventArch");
        this.producerEventArchService.addProducerEventArch(producerEvent);
      } catch (Exception e) {
        this.LOGGER.error("producer eventArch create fail", e);
      } 
    } 
  }
  
  public void reSentProducerEvent(Long eventId) {
    List<ProducerEvent> select = this.producerEventRepository.select("eventId", eventId);
    if (CollectionUtils.isEmpty(select)) {
      this.LOGGER.error("can not find  producer event by id : {0}", eventId);
      return;
    } 
    sentProducerEvent(select.get(0));
  }
  
  private ProducerConfigDTO getProducerConfig(String tableName) {
    String redisKey = "hdtt:config:" + this.serviceName + ":" + tableName;
    String hashKey = "producer";
    String jsonString = this.redisHelper.hshGet(redisKey, hashKey);
    if (StringUtils.isEmpty(jsonString))
      return null; 
    ProducerConfigDTO producerConfigDTO = this.redisHelper.fromJson(jsonString, ProducerConfigDTO.class);
    return producerConfigDTO;
  }
}
